package com.homedo.async.mysql.support.repository;

import com.homedo.async.common.dto.Pagination;
import com.homedo.async.mysql.support.dataobject.Example;
import com.homedo.async.mysql.support.mapper.delete.DeleteMapperExampleI;
import com.homedo.async.mysql.support.mapper.delete.DeleteMapperExampleImpl;
import com.homedo.async.mysql.support.mapper.delete.DeleteMapperI;
import com.homedo.async.mysql.support.mapper.delete.DeleteMapperImpl;
import com.homedo.async.mysql.support.mapper.insert.SaveMapperExampleI;
import com.homedo.async.mysql.support.mapper.insert.SaveMapperExampleImpl;
import com.homedo.async.mysql.support.mapper.insert.SaveMapperI;
import com.homedo.async.mysql.support.mapper.insert.SaveMapperImpl;
import com.homedo.async.mysql.support.mapper.select.*;
import com.homedo.async.mysql.support.mapper.update.UpdateMapperExampleI;
import com.homedo.async.mysql.support.mapper.update.UpdateMapperExampleImpl;
import com.homedo.async.mysql.support.mapper.update.UpdateMapperI;
import com.homedo.async.mysql.support.mapper.update.UpdateMapperImpl;
import com.homedo.async.mysql.support.meta.EntityMetaData;
import com.homedo.async.mysql.support.meta.EntityMetaDataCache;
import com.homedo.plainly.logger.Logger;
import com.homedo.plainly.logger.LoggerFactory;
import io.vertx.sqlclient.Pool;
import io.vertx.sqlclient.SqlConnection;
import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;

import javax.annotation.PostConstruct;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;

/**
 * @author jgshun
 * @date 2020/3/5 4:35 PM
 */
public class ReactiveMysqlRepository<T, ID> implements ReactiveMysqlRepositoryI<T, ID> {
    private static final Logger _LOGGER = LoggerFactory.getLogger(ReactiveMysqlRepository.class);

    @Autowired
    protected Pool pool;

    private EntityMetaData<T> entityMetaData;

    private SaveMapperI<T, ID> saveMapper;
    private UpdateMapperI<T, ID> updateMapper;
    private DeleteMapperI<T, ID> deleteMapper;
    private SelectMapperI<T, ID> selectMapper;
    private SelectPaginationMapperI<T, ID> selectPaginationMapper;
    private SaveMapperExampleI<T, ID> saveMapperExample;
    private UpdateMapperExampleI<T, ID> updateMapperExample;
    private DeleteMapperExampleI<T, ID> deleteMapperExample;
    private SelectMapperExampleI<T, ID> selectMapperExample;

    public ReactiveMysqlRepository() {
        Class<T> entityClz = null;
        Type type3 = this.getClass().getGenericSuperclass();
        if (type3 instanceof ParameterizedType) {
            entityClz = (Class<T>) ((ParameterizedType) type3).getActualTypeArguments()[0];
        } else {
            throw new RuntimeException("[初始化元数据失败:" + this.getClass() + "]");
        }

        this.entityMetaData = EntityMetaDataCache.get(entityClz);
    }

    @PostConstruct
    void afterPropertiesSet() throws Exception {
        this.saveMapper = new SaveMapperImpl<>(this.pool, this.entityMetaData);
        this.updateMapper = new UpdateMapperImpl<>(this.pool, this.entityMetaData);
        this.deleteMapper = new DeleteMapperImpl<>(this.pool, this.entityMetaData);
        this.selectMapper = new SelectMapperImpl<>(this.pool, this.entityMetaData);
        this.selectPaginationMapper = new SelectPaginationMapperImpl<>(this.pool, this.entityMetaData);
        this.saveMapperExample = new SaveMapperExampleImpl<>(this.pool, this.entityMetaData);
        this.updateMapperExample = new UpdateMapperExampleImpl<>(this.pool, this.entityMetaData);
        this.deleteMapperExample = new DeleteMapperExampleImpl<>(this.pool, this.entityMetaData);
        this.selectMapperExample = new SelectMapperExampleImpl<>(this.pool, this.entityMetaData);

    }

    @Override
    public <S extends T> Mono<S> saveSelective(S entity) {
        return this.saveMapper.saveSelective(entity);
    }

    @Override
    public <S extends T> Flux<S> saveSelective(Iterable<S> entities) {
        return this.saveMapper.saveSelective(entities);
    }

    @Override
    public <S extends T> Flux<S> saveSelective(Publisher<S> entityStream) {
        return this.saveMapper.saveSelective(entityStream);
    }

    @Override
    public <S extends T> Mono<S> save(S entity) {
        return this.saveMapper.save(entity);
    }

    @Override
    public <S extends T> Flux<S> save(Iterable<S> entities) {
        return this.saveMapper.save(entities);
    }

    @Override
    public <S extends T> Flux<S> save(Publisher<S> entityStream) {
        return this.saveMapper.save(entityStream);
    }

    @Override
    public <S extends T> Mono<S> saveSelective(S entity, SqlConnection connection) {
        return this.saveMapper.saveSelective(entity, connection);
    }

    @Override
    public <S extends T> Flux<S> saveSelective(Iterable<S> entities, SqlConnection connection) {
        return this.saveMapper.saveSelective(entities, connection);
    }

    @Override
    public <S extends T> Flux<S> saveSelective(Publisher<S> entityStream, SqlConnection connection) {
        return this.saveMapper.saveSelective(entityStream, connection);
    }

    @Override
    public <S extends T> Mono<S> save(S entity, SqlConnection connection) {
        return this.saveMapper.save(entity, connection);
    }

    @Override
    public <S extends T> Flux<S> save(Iterable<S> entities, SqlConnection connection) {
        return this.saveMapper.save(entities, connection);
    }

    @Override
    public <S extends T> Flux<S> save(Publisher<S> entityStream, SqlConnection connection) {
        return this.saveMapper.save(entityStream, connection);
    }

    @Override
    public Mono<T> findById(ID id) {
        return this.selectMapper.findById(id);
    }

    @Override
    public Mono<T> findById(Publisher<ID> publisher) {
        return this.selectMapper.findById(publisher);
    }

    @Override
    public Mono<Boolean> existsById(ID id) {
        return this.selectMapper.existsById(id);
    }

    @Override
    public Mono<Boolean> existsById(Publisher<ID> publisher) {
        return this.selectMapper.existsById(publisher);
    }

    @Override
    public Flux<T> findAll() {
        return this.selectMapper.findAll();
    }

    @Override
    public Flux<T> findByRecordSelective(T entity) {
        return this.selectMapper.findByRecordSelective(entity);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream) {
        return this.selectMapper.findByRecordSelective(entityStream);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities) {
        return this.selectMapper.findByRecordsSelective(entities);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream) {
        return this.selectMapper.findByRecordsSelective(entityStream);
    }

    @Override
    public Flux<T> findByRecord(T entity) {
        return this.selectMapper.findByRecord(entity);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream) {
        return this.selectMapper.findByRecord(entityStream);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities) {
        return this.selectMapper.findByRecords(entities);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream) {
        return this.selectMapper.findByRecords(entityStream);
    }

    @Override
    public Flux<T> findByIds(Iterable<ID> ids) {
        return this.selectMapper.findByIds(ids);
    }

    @Override
    public Flux<T> findByIds(Publisher<ID> idStream) {
        return this.selectMapper.findByIds(idStream);
    }

    @Override
    public Mono<Long> countAll() {
        return this.selectMapper.countAll();
    }

    @Override
    public Mono<Long> countByRecordSelective(T entity) {
        return this.selectMapper.countByRecordSelective(entity);
    }

    @Override
    public Mono<Long> countByRecordSelective(Publisher<? extends T> publisher) {
        return this.selectMapper.countByRecordSelective(publisher);
    }

    @Override
    public Mono<Long> countByRecord(T entity) {
        return this.selectMapper.countByRecord(entity);
    }

    @Override
    public Mono<Long> countByRecord(Publisher<? extends T> publisher) {
        return this.selectMapper.countByRecord(publisher);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Iterable<? extends T> entities) {
        return this.selectMapper.countByRecordsSelective(entities);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Publisher<? extends T> entityStream) {
        return this.selectMapper.countByRecordsSelective(entityStream);
    }

    @Override
    public Mono<Long> countByRecords(Iterable<? extends T> entities) {
        return this.selectMapper.countByRecords(entities);
    }

    @Override
    public Mono<Long> countByRecords(Publisher<? extends T> entityStream) {
        return this.selectMapper.countByRecords(entityStream);
    }

    @Override
    public Mono<T> findById(ID id, SqlConnection connection) {
        return this.selectMapper.findById(id, connection);
    }

    @Override
    public Mono<T> findById(Publisher<ID> id, SqlConnection connection) {
        return this.selectMapper.findById(id, connection);
    }

    @Override
    public Mono<Boolean> existsById(ID id, SqlConnection connection) {
        return this.selectMapper.existsById(id, connection);
    }

    @Override
    public Mono<Boolean> existsById(Publisher<ID> id, SqlConnection connection) {
        return this.selectMapper.existsById(id, connection);
    }

    @Override
    public Flux<T> findAll(SqlConnection connection) {
        return this.selectMapper.findAll(connection);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, SqlConnection connection) {
        return this.selectMapper.findByRecordSelective(entity, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.findByRecordSelective(entityStream, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.selectMapper.findByRecordsSelective(entities, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.findByRecordsSelective(entityStream, connection);
    }

    @Override
    public Flux<T> findByRecord(T entity, SqlConnection connection) {
        return this.selectMapper.findByRecord(entity, connection);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.findByRecord(entityStream, connection);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, SqlConnection connection) {
        return this.selectMapper.findByRecords(entities, connection);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.findByRecords(entityStream, connection);
    }

    @Override
    public Flux<T> findByIds(Iterable<ID> ids, SqlConnection connection) {
        return this.selectMapper.findByIds(ids, connection);
    }

    @Override
    public Flux<T> findByIds(Publisher<ID> idStream, SqlConnection connection) {
        return this.selectMapper.findByIds(idStream, connection);
    }

    @Override
    public Mono<Long> countAll(SqlConnection connection) {
        return this.selectMapper.countAll(connection);
    }

    @Override
    public Mono<Long> countByRecordSelective(T entity, SqlConnection connection) {
        return this.selectMapper.countByRecordSelective(entity, connection);
    }

    @Override
    public Mono<Long> countByRecordSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.countByRecordSelective(entityStream, connection);
    }

    @Override
    public Mono<Long> countByRecord(T entity, SqlConnection connection) {
        return this.selectMapper.countByRecord(entity, connection);
    }

    @Override
    public Mono<Long> countByRecord(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.countByRecord(entityStream, connection);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.selectMapper.countByRecordsSelective(entities, connection);
    }

    @Override
    public Mono<Long> countByRecordsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.countByRecordsSelective(entityStream, connection);
    }

    @Override
    public Mono<Long> countByRecords(Iterable<? extends T> entities, SqlConnection connection) {
        return this.selectMapper.countByRecords(entities, connection);
    }

    @Override
    public Mono<Long> countByRecords(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.selectMapper.countByRecords(entityStream, connection);
    }

    @Override
    public Flux<T> findAll(Pagination<T> pagination) {
        return this.selectPaginationMapper.findAll(pagination);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecordSelective(entity, pagination);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecordSelective(entityStream, pagination);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecordsSelective(entities, pagination);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecordsSelective(entityStream, pagination);
    }

    @Override
    public Flux<T> findByRecord(T entity, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecord(entity, pagination);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecord(entityStream, pagination);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecords(entities, pagination);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.selectPaginationMapper.findByRecords(entityStream, pagination);
    }

    @Override
    public Flux<T> findAll(Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findAll(pagination, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(T entity, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecordSelective(entity, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordSelective(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecordSelective(entityStream, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecordsSelective(entities, pagination, connection);
    }

    @Override
    public Flux<T> findByRecordsSelective(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecordsSelective(entityStream, pagination, connection);
    }

    @Override
    public Flux<T> findByRecord(T entity, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecord(entity, pagination, connection);
    }

    @Override
    public Flux<T> findByRecord(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecord(entityStream, pagination, connection);
    }

    @Override
    public Flux<T> findByRecords(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecords(entities, pagination, connection);
    }

    @Override
    public Flux<T> findByRecords(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return this.selectPaginationMapper.findByRecords(entityStream, pagination, connection);
    }

    @Override
    public Mono<Long> deleteById(ID id) {
        return this.deleteMapper.deleteById(id);
    }

    @Override
    public Mono<Long> deleteById(Publisher<ID> publisher) {
        return this.deleteMapper.deleteById(publisher);
    }

    @Override
    public Mono<Long> deleteByIds(Iterable<ID> ids) {
        return this.deleteMapper.deleteByIds(ids);
    }

    @Override
    public Mono<Long> deleteByIds(Publisher<ID> idStream) {
        return this.deleteMapper.deleteByIds(idStream);
    }

    @Override
    public Mono<Long> deleteByRecordSelective(T entity) {
        return this.deleteMapper.deleteByRecordSelective(entity);
    }

    @Override
    public Mono<Long> deleteByRecordsSelective(Iterable<? extends T> entities) {
        return this.deleteMapper.deleteByRecordsSelective(entities);
    }

    @Override
    public Mono<Long> deleteByRecordsSelective(Publisher<? extends T> entityStream) {
        return this.deleteMapper.deleteByRecordsSelective(entityStream);
    }

    @Override
    public Mono<Long> deleteByRecord(T entity) {
        return this.deleteMapper.deleteByRecord(entity);
    }

    @Override
    public Mono<Long> deleteByRecords(Iterable<? extends T> entities) {
        return this.deleteMapper.deleteByRecords(entities);
    }

    @Override
    public Mono<Long> deleteByRecords(Publisher<? extends T> entityStream) {
        return this.deleteMapper.deleteByRecords(entityStream);
    }

    @Override
    public Mono<Long> deleteAll() {
        return this.deleteMapper.deleteAll();
    }

    @Override
    public Mono<Long> deleteById(ID id, SqlConnection connection) {
        return this.deleteMapper.deleteById(id, connection);
    }

    @Override
    public Mono<Long> deleteById(Publisher<ID> id, SqlConnection connection) {
        return this.deleteMapper.deleteById(id, connection);
    }

    @Override
    public Mono<Long> deleteByIds(Iterable<ID> ids, SqlConnection connection) {
        return this.deleteMapper.deleteByIds(ids, connection);
    }

    @Override
    public Mono<Long> deleteByIds(Publisher<ID> idStream, SqlConnection connection) {
        return this.deleteMapper.deleteByIds(idStream, connection);
    }

    @Override
    public Mono<Long> deleteByRecordSelective(T entity, SqlConnection connection) {
        return this.deleteMapper.deleteByRecordSelective(entity, connection);
    }

    @Override
    public Mono<Long> deleteByRecordsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.deleteMapper.deleteByRecordsSelective(entities, connection);
    }

    @Override
    public Mono<Long> deleteByRecordsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.deleteMapper.deleteByRecordsSelective(entityStream, connection);
    }

    @Override
    public Mono<Long> deleteByRecord(T entity, SqlConnection connection) {
        return this.deleteMapper.deleteByRecord(entity, connection);
    }

    @Override
    public Mono<Long> deleteByRecords(Iterable<? extends T> entities, SqlConnection connection) {
        return this.deleteMapper.deleteByRecords(entities, connection);
    }

    @Override
    public Mono<Long> deleteByRecords(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.deleteMapper.deleteByRecords(entityStream, connection);
    }

    @Override
    public Mono<Long> deleteAll(SqlConnection connection) {
        return this.deleteMapper.deleteAll(connection);
    }

    @Override
    public Mono<Long> updateByIdSelective(T entity) {
        return this.updateMapper.updateByIdSelective(entity);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Iterable<? extends T> entities) {
        return this.updateMapper.updateByIdsSelective(entities);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Publisher<? extends T> entityStream) {
        return this.updateMapper.updateByIdsSelective(entityStream);
    }

    @Override
    public Mono<Long> updateById(T entity) {
        return this.updateMapper.updateById(entity);
    }

    @Override
    public Mono<Long> updateByIds(Iterable<? extends T> entities) {
        return this.updateMapper.updateByIds(entities);
    }

    @Override
    public Mono<Long> updateByIds(Publisher<? extends T> entityStream) {
        return this.updateMapper.updateByIds(entityStream);
    }

    @Override
    public Mono<Long> updateByIdSelective(T entity, SqlConnection connection) {
        return this.updateMapper.updateByIdSelective(entity, connection);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Iterable<? extends T> entities, SqlConnection connection) {
        return this.updateMapper.updateByIdsSelective(entities, connection);
    }

    @Override
    public Mono<Long> updateByIdsSelective(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.updateMapper.updateByIdsSelective(entityStream, connection);
    }

    @Override
    public Mono<Long> updateById(T entity, SqlConnection connection) {
        return this.updateMapper.updateById(entity, connection);
    }

    @Override
    public Mono<Long> updateByIds(Iterable<? extends T> entities, SqlConnection connection) {
        return this.updateMapper.updateByIds(entities, connection);
    }

    @Override
    public Mono<Long> updateByIds(Publisher<? extends T> entityStream, SqlConnection connection) {
        return this.updateMapper.updateByIds(entityStream, connection);
    }

    @Override
    public Mono<Pagination<T>> findAll2Pagination(Pagination<T> pagination) {
        return this.findAll2Pagination(pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecordSelective2Pagination(T entity, Pagination<T> pagination) {
        return this.findByRecordSelective2Pagination(entity, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecordSelective2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecordSelective2Pagination(entityStream, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecordsSelective2Pagination(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.findByRecordsSelective2Pagination(entities, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecordsSelective2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecordsSelective2Pagination(entityStream, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecord2Pagination(T entity, Pagination<T> pagination) {
        return this.findByRecord2Pagination(entity, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecord2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecord2Pagination(entityStream, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecords2Pagination(Iterable<? extends T> entities, Pagination<T> pagination) {
        return this.findByRecords2Pagination(entities, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findByRecords2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination) {
        return this.findByRecords2Pagination(entityStream, pagination, null);
    }

    @Override
    public Mono<Pagination<T>> findAll2Pagination(Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> allMono = Mono.create(allSink -> {

                this.findAll(pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countAll(connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, allMono, countMono, sink);

            } else {
                allMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecordSelective2Pagination(T entity, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecordSelective(entity, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecordSelective(entity, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecordSelective2Pagination(Publisher<? extends T> entity, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecordSelective(entity, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecordSelective(entity, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecordsSelective2Pagination(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecordsSelective(entities, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecordsSelective(entities, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecordsSelective2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecordsSelective(entityStream, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecordsSelective(entityStream, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecord2Pagination(T entity, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecord(entity, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecord(entity, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecord2Pagination(Publisher<? extends T> entity, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecord(entity, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecord(entity, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecords2Pagination(Iterable<? extends T> entities, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecords(entities, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecords(entities, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }

    @Override
    public Mono<Pagination<T>> findByRecords2Pagination(Publisher<? extends T> entityStream, Pagination<T> pagination, SqlConnection connection) {
        return Mono.<Pagination<T>>create(sink -> {

            Mono<Object> findMono = Mono.create(allSink -> {

                this.findByRecords(entityStream, pagination, connection).buffer().subscribe(e -> {
                    allSink.success(e);
                }, error -> allSink.error(error), () -> allSink.success());

            }).subscribeOn(Schedulers.elastic());

            if (pagination.isCounted()) {
                Mono<Long> countMono = this.countByRecords(entityStream, connection).subscribeOn(Schedulers.elastic());

                this.merge2Pagination(pagination, findMono, countMono, sink);

            } else {
                findMono.doOnSuccess(e -> {
                    pagination.setItems((List) e);
                    sink.success(pagination);
                }).doOnError(error -> sink.error(error)).subscribe();
            }
        });
    }


    private void merge2Pagination(Pagination<T> pagination, Mono<Object> findMono, Mono<Long> countMono, MonoSink<Pagination<T>> sink) {
        Flux.merge(findMono, countMono).subscribe(e -> {
            if (e instanceof Long) {
                pagination.setCount(Long.valueOf(e.toString()).intValue());
            } else {
                pagination.setItems((List) e);
            }
        }, error -> sink.error(error), new Runnable() {
            @Override
            public void run() {
                sink.success(pagination);
            }
        });
    }

    @Override
    public Mono<Long> delete(Example example) {
        return this.deleteMapperExample.delete(example);
    }

    @Override
    public Mono<Long> delete(Example example, SqlConnection connection) {
        return this.deleteMapperExample.delete(example, connection);
    }

    @Override
    public Mono<Long> save(Example example) {
        return this.saveMapperExample.save(example);
    }

    @Override
    public Mono<Long> save(Example example, SqlConnection connection) {
        return this.saveMapperExample.save(example, connection);
    }

    @Override
    public Flux<T> findByExample(Example example) {
        return this.selectMapperExample.findByExample(example);
    }

    @Override
    public Flux<T> findByExample(Example example, SqlConnection connection) {
        return this.selectMapperExample.findByExample(example, connection);
    }

    @Override
    public Flux<T> findByExample(Example example, Pagination<T> pagination) {
        return this.selectMapperExample.findByExample(example, pagination);
    }

    @Override
    public Flux<T> findByExample(Example example, Pagination<T> pagination, SqlConnection connection) {
        return this.selectMapperExample.findByExample(example, pagination, connection);
    }

    @Override
    public Mono<Long> update(Example example) {
        return this.updateMapperExample.update(example);
    }

    @Override
    public Mono<Long> update(Example example, SqlConnection connection) {
        return this.updateMapperExample.update(example, connection);
    }
}
